package com.skew

import org.apache.spark.{SparkConf, SparkContext}

import scala.util.Random

/**
  *
  * @author ymy.hadoop
  *
  */
object Demo01 {

  def main(args: Array[String]): Unit = {

      val conf = new SparkConf().setAppName("Demo").setMaster("local[*]")
      val sc = new SparkContext(conf)

      val array = new Array[Int](10000000)
      for (i <- 0 to 9999999){
        array(i) = new Random().nextInt(10)
      }

      val rdd = sc.parallelize(array)
     // 取样操作
     // val num = rdd.sample(true,0.9,0).count()

     val mapprdd = rdd.map((_,1))
     // 未处理数据倾斜前
     mapprdd.countByKey().foreach(println)

     // 处理数据倾斜后
//    val prifixrdd = mapprdd.map(x=>{
//      val prifix = new Random().nextInt(10)
//      (prifix + "_" + x._1,x._2)
//    })
//
//     // 加上随机前缀的key进行局部聚合
//     val tmprdd = prifixrdd.reduceByKey(_+_)
//     // 去除随机前缀
//     val newrdd = tmprdd.map(x=>((x._1).split("_")(1),x._2))
//     // 最终聚合
//     newrdd.reduceByKey(_+_).foreach(println)

      sc.stop()
  }
}
